package com.nepu.spark.realtime.util

import java.time.LocalDate
import java.util

import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.serializer.SerializeConfig
import org.apache.http.HttpHost
import org.elasticsearch.action.bulk.BulkRequest
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.action.search.{SearchRequest, SearchResponse}
import org.elasticsearch.client.indices.GetIndexRequest
import org.elasticsearch.client.{RequestOptions, RestClient, RestClientBuilder, RestHighLevelClient}
import org.elasticsearch.cluster.metadata.Metadata.XContentContext
import org.elasticsearch.common.xcontent.XContentType
import org.elasticsearch.search.SearchHit
import org.elasticsearch.search.builder.SearchSourceBuilder

import scala.collection.mutable.ListBuffer

/**
  * 这个是es的工具类
  *
  * @author chenshuaijun
  * @create 2022-11-04 20:13
  */
object MyElasticSearchUtils {

  val client: RestHighLevelClient = build()

  /**
    * 创建客户端
    *
    * @return 返回客户端
    */
  def build(): RestHighLevelClient = {
    val clientBuilder: RestClientBuilder = RestClient.builder(new HttpHost("hadoop102", 9200))
    val highLevelClient = new RestHighLevelClient(clientBuilder)
    highLevelClient
  }

  /**
    * 关闭客户端
    */
  def close(): Unit = {
    if (client != null) {
      client.close()
    }
  }

  /**
    * 将数据批量的写入
    *
    * @param indexName 索引的名字
    * @param documents (id,文档)
    */
  def batchIdempotentWrite(indexName: String, documents: List[(String, AnyRef)]): Unit = {
    val bulkRequest = new BulkRequest()
    for ((docId, docValue) <- documents) {
      val request = new IndexRequest(indexName)
      val docValueJson: String = JSON.toJSONString(docValue, new SerializeConfig(true))
      request.id(docId)
      request.source(docValueJson, XContentType.JSON)
      bulkRequest.add(request)
    }
    client.bulk(bulkRequest, RequestOptions.DEFAULT)
  }

  /**
    * 根据索引和属性，从es中查询对应的属性
    *
    * @param indexName 索引名称
    * @param fieldName 属性名称
    * @return
    */
  def searchField(indexName: String, fieldName: String): List[String] = {
    // 判断索引是否存在
    val indexRequest = new GetIndexRequest(indexName)
    val isExist: Boolean = client.indices().exists(indexRequest, RequestOptions.DEFAULT)
    if (!isExist) {
      // 如果当前的索引都不存在，就直接的返回了
      return null
    }

    val fieldBuffer = new ListBuffer[String]
    val searchRequest = new SearchRequest(indexName)
    val searchSourceBuilder = new SearchSourceBuilder()
    searchSourceBuilder.fetchSource(fieldName, null).size(100000)
    searchRequest.source(searchSourceBuilder)
    val searchResponse: SearchResponse = client.search(searchRequest, RequestOptions.DEFAULT)
    // 对结果进行处理
    val hits: Array[SearchHit] = searchResponse.getHits.getHits
    for (hit <- hits) {
      val sourceMap: util.Map[String, AnyRef] = hit.getSourceAsMap
      val field: String = sourceMap.get(fieldName).toString
      fieldBuffer.append(field)
    }
    fieldBuffer.toList
  }

}
